package com.atliuzu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atliuzu.bean.UserLoginBean;
import com.atliuzu.utils.ClickHouseUtil;
import com.atliuzu.utils.DateFormatUtil;
import com.atliuzu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.codehaus.janino.IClass;

import java.time.Duration;

/**
 * @author w
 * @create 2022-08-20-10:22
 */
public class DwsUserUserBackWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment ( );
        env.setParallelism ( 1 );

        //TODO 2.状态后端设置
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                3, Time.days(1), Time.minutes(1)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage(
//                "hdfs://hadoop102:8020/ck"
//        );
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //TODO 3.读取DWD层页面日志数据
        String pageTopic = "dwd_traffic_page_log";
        String groupId = "user_login_0212";
        DataStreamSource<String> kafkaSourceDS = env.addSource ( MyKafkaUtil.getFlinkKafkaConsumer ( pageTopic, groupId ) );

        //TODO 4.转换为JSON对象并过滤数据
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaSourceDS.flatMap ( new FlatMapFunction<String, JSONObject> ( ) {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject ( value );
                String uid = jsonObject.getJSONObject ( "common" ).getString ( "uid" );
                String lastPageId = jsonObject.getJSONObject ( "page" ).getString ( "last_page_id" );
                if (uid != null && (lastPageId == null || "login".equals ( lastPageId ))) {
                    out.collect ( jsonObject );
                }
            }
        } );
        jsonObjDS.print ( "jsonObjDS >>>>" );

        //TODO 5.提取时间戳生成watermark
        SingleOutputStreamOperator<JSONObject> jsonObjWithWMDS = jsonObjDS.assignTimestampsAndWatermarks ( WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness ( Duration.ofSeconds ( 2 ) ).withTimestampAssigner ( new SerializableTimestampAssigner<JSONObject> ( ) {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                return element.getLong ( "ts" );
            }
        } ) );

        //TODO 6.按照uid分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjWithWMDS.keyBy ( json -> json.getJSONObject ( "common" ).getString ( "uid" ) );

        //TODO 7.使用状态编程进行去重&转为JavaBean
        SingleOutputStreamOperator<UserLoginBean> userLoginDS = keyedStream.flatMap ( new RichFlatMapFunction<JSONObject, UserLoginBean> ( ) {
            //定义状态
            private ValueState<String> lastValueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                lastValueState = getRuntimeContext ( ).getState ( new ValueStateDescriptor<String> ( "last-state", String.class ) );
            }

            @Override
            public void flatMap(JSONObject value, Collector<UserLoginBean> out) throws Exception {
                //获取状态当前日期以及当前数据的日期
                String lastDt = lastValueState.value ( );
                Long currentTs = value.getLong ( "ts" );
                String currentDt = DateFormatUtil.toDate ( currentTs );

                //初始化回流用户数
                long backCt = 0L;
                if (lastDt == null) {
                    lastValueState.update ( currentDt );
                } else if (!lastDt.equals ( currentDt )) {
                    if ((currentTs - DateFormatUtil.toTs ( lastDt, false ))/ (24 * 60 * 60 * 1000L) > 7) {
                        backCt = 1;
                    }
                    lastValueState.update ( currentDt );
                }
                if (backCt == 1L) {
                    out.collect ( new UserLoginBean ( "", "", backCt, null ) );
                }

            }
        } );

        //TODO 8.开窗聚合
        SingleOutputStreamOperator<UserLoginBean> reduceDS = userLoginDS.windowAll ( TumblingEventTimeWindows.of ( Time.seconds ( 10 ) ) ).reduce ( new ReduceFunction<UserLoginBean> ( ) {
            @Override
            public UserLoginBean reduce(UserLoginBean value1, UserLoginBean value2) throws Exception {
                value1.setBackCt ( value1.getBackCt ( ) + value2.getBackCt ( ) );
                return value1;
            }
        }, new AllWindowFunction<UserLoginBean, UserLoginBean, TimeWindow> ( ) {
            @Override
            public void apply(TimeWindow window, Iterable<UserLoginBean> values, Collector<UserLoginBean> out) throws Exception {
                //获取数据
                UserLoginBean next = values.iterator ( ).next ( );
                //补充信息
                next.setTs ( System.currentTimeMillis ( ) );
                next.setStt ( DateFormatUtil.toYmdHms ( window.getStart ( ) ) );
                next.setEdt ( DateFormatUtil.toYmdHms ( window.getEnd ( ) ) );

                //输出数据
                out.collect ( next );
            }
        } );

        reduceDS.print ( "reduceDS >>>>>" );

        //TODO 9.将数据写出到clickHouse
        reduceDS.addSink ( ClickHouseUtil.getJdbcSink ( "insert into dws_user_back_window values(?,?,?,?)" ) );

        //TODO 10.启动任务
        env.execute ("DwsUserUserBackWindow");
    }
}
